package spring.cloud.tasks.tasks_manager.service.impl;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spring.cloud.tasks.tasks_manager.database.entity.NameSpaceZooKeeperClusterMappingDto;
import spring.cloud.tasks.tasks_manager.database.entity.ZooKeeperClusterDto;
import spring.cloud.tasks.tasks_manager.database.service.NameSpaceDtoService;
import spring.cloud.tasks.tasks_manager.database.service.NameSpaceZooKeeperClusterMappingDtoService;
import spring.cloud.tasks.tasks_manager.database.service.ZooKeeperClusterDtoService;
import spring.cloud.tasks.tasks_manager.domain.NameSpaceClient;
import spring.cloud.tasks.tasks_manager.domain.NameSpaceConfiguration;
import spring.cloud.tasks.tasks_manager.domain.ZooKeeperCluster;
import spring.cloud.tasks.tasks_manager.exception.ConsoleException;
import spring.cloud.tasks.tasks_manager.service.ZooKeeperService;
import spring.cloud.tasks.tasks_manager.tasks_manager.TaskManager;
import spring.cloud.tasks.tasks_manager.util.*;
import spring.cloud.tasks.tasks_manager.zookeeper.ZooKeeper;
import spring.cloud.tasks.tasks_manager.zookeeper.ZooKeeperRepository;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ZooKeeperServiceImpl implements ZooKeeperService {


    protected static final String NAMESPACE_CREATOR_NAME = "REST_API";


    private static final Logger log = LoggerFactory.getLogger(ZooKeeperServiceImpl.class);

    // namespace is unique in all zkClusters
    private final ConcurrentHashMap<String /** nns */, NameSpaceClient> nameSpaceToNameSpaceClientMap = new ConcurrentHashMap<String /** nns */, NameSpaceClient>();
    private final ConcurrentHashMap<String, Object> nameSpaceLock = new ConcurrentHashMap<String, Object>(); // maybe could remove in right time
    // namespace is unique in all zkClusters
    private final ConcurrentHashMap<String /** nns **/, TaskManager> nameSpaceToTaskManagerMap = new ConcurrentHashMap<String /** nns **/, TaskManager>();
    private final Set<String> restrictComputeZkClusterKeys = Sets.newHashSet();
    @Resource
    private ZooKeeperRepository zooKeeperRepository;

    //    @Resource
//    private UpdateTaskConfigService updateTaskConfigService;
    @Resource
    private ZooKeeperClusterDtoService zooKeeperClusterDtoService;

    @Resource
    private NameSpaceZooKeeperClusterMappingDtoService nameSpaceZooKeeperClusterMappingDtoService;
    @Resource
    private NameSpaceDtoService nameSpaceDtoService;
    /**
     * 为保证values有序
     **/
    private LinkedHashMap<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = new LinkedHashMap<>();
    private List<String> onlineNameSpaceList = new ArrayList<String>();

    private Timer localRefreshTimer = null;

    private Timer localRefreshIfNecessaryTimer = null;

    private ExecutorService localRefreshThreadPool = null;

    public void init() {

        localRefresh();
        initLocalRefreshThreadPool();
        startLocalRefreshTimer();
        startLocalRefreshIfNecessaryTimer();
    }


    @PreDestroy
    public void destroy() {
        Iterator<Entry<String, ZooKeeperCluster>> iterator = zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            closeZooKeeperClusterClient(iterator.next().getValue());
        }
        if (localRefreshThreadPool != null) {
            localRefreshThreadPool.shutdownNow();
        }
        if (localRefreshTimer != null) {
            localRefreshTimer.cancel();
        }
        if (localRefreshIfNecessaryTimer != null) {
            localRefreshIfNecessaryTimer.cancel();
        }

    }

    private synchronized void localRefresh() {
        try {
            log.info("Start refresh RegCenter");
            long startTime = System.currentTimeMillis();
            refreshRestrictComputeZooKeeperClusters();
            if (restrictComputeZkClusterKeys.isEmpty()) {
                log.warn("根据Console的集群ID:" + "" + ",找不到配置可以参与Sharding和Dashboard计算的zk集群");
                return;
            }
            refresh();

            refreshNameSpaceToTaskManagerMap();
            log.info("End refresh RegCenter, cost {}ms", System.currentTimeMillis() - startTime);
        } catch (Exception e) {
            log.error("refresh RegCenter error", e);
        }
    }

    /**
     * 解析Console集群和zk的映射关系 数据库中配置的例子如下： CONSOLE-1:/saturn,/forVdos;CONSOLE-2:/zk3; 如果不存在此配置项，则可以计算所有zk集群；
     */
    private void refreshRestrictComputeZooKeeperClusters() throws ConsoleException {
        // clear 当前可计算的zkCluster集群列表
        restrictComputeZkClusterKeys.clear();

        restrictComputeZkClusterKeys.addAll(getZooKeeperClusterKeyList());
        return;
    }

    /**
     * 刷新ZooKeeperCluster/NameSpace/调度器信息
     */
    private void refresh() {
        // 获取新的ZooKeeperCluster
        Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = getZooKeeperClusterKeyToZooKeeperClusterMap();
        // 对比旧的。
        // 1不包含的，关闭操作；
        // 2包含的，检查属性是否相同，如果相同，则直接赋值，否则，关闭旧的
        // 3其他直接关闭
        closeInvalidZooKeeperClusterClient(zooKeeperClusterKeyToZooKeeperClusterMap);
        // 完善curatorFramework。如果没有，则新建
        connectZooKeeperCluster(zooKeeperClusterKeyToZooKeeperClusterMap);
        List<String> onlineNameSpaceList = new ArrayList<String>();
        // 完善ZkCluster中的注册中心信息，关闭迁移了的域，新建迁移过来的域
        for (Map.Entry<String, ZooKeeperCluster> entry : zooKeeperClusterKeyToZooKeeperClusterMap.entrySet()) {
            String zooKeeperClusterKey = entry.getKey();
            ZooKeeperCluster zooKeeperCluster = entry.getValue();
            //
            List<NameSpaceZooKeeperClusterMappingDto> nameSpaceZooKeeperClusterMappingDtoList = nameSpaceZooKeeperClusterMappingDtoService.getNameSpaceZooKeeperClusterMappingDtoList(zooKeeperClusterKey);
            List<NameSpaceConfiguration> nameSpaceConfigurationList = zooKeeperCluster.getNameSpaceConfigurationList();
            closeMigrationNameSpace(zooKeeperClusterKey, nameSpaceZooKeeperClusterMappingDtoList, nameSpaceConfigurationList);
            initNameSpace(onlineNameSpaceList, zooKeeperCluster, nameSpaceZooKeeperClusterMappingDtoList, nameSpaceConfigurationList);
        }
        // 直接赋值新的
        this.zooKeeperClusterKeyToZooKeeperClusterMap = (LinkedHashMap<String, ZooKeeperCluster>) zooKeeperClusterKeyToZooKeeperClusterMap;
        log.info("new zkClusterMap is : {}", this.zooKeeperClusterKeyToZooKeeperClusterMap);
        this.onlineNameSpaceList = onlineNameSpaceList;
    }

    /**
     * zooKeeperClusterKeyToZooKeeperClusterMap
     *
     * @param newZooKeeperClusterKeyToZooKeeperClusterMap
     */
    private void closeInvalidZooKeeperClusterClient(Map<String, ZooKeeperCluster> newZooKeeperClusterKeyToZooKeeperClusterMap) {
        Iterator<Entry<String, ZooKeeperCluster>> iterator = this.zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, ZooKeeperCluster> next = iterator.next();
            String zooKeeperClusterKey = next.getKey();
            ZooKeeperCluster zooKeeperCluster = next.getValue();
            //
            if (!newZooKeeperClusterKeyToZooKeeperClusterMap.containsKey(zooKeeperClusterKey)) {
                iterator.remove();
                closeZooKeeperClusterClient(zooKeeperCluster);
            } else {
                ZooKeeperCluster newZooKeeperCluster = newZooKeeperClusterKeyToZooKeeperClusterMap.get(zooKeeperClusterKey);
                if (zooKeeperCluster.equals(newZooKeeperCluster)) {
                    newZooKeeperClusterKeyToZooKeeperClusterMap.put(zooKeeperClusterKey, zooKeeperCluster);
                } else if (zooKeeperCluster.equalsNoNeedReconnect(newZooKeeperCluster)) {
                    zooKeeperCluster.setDescription(newZooKeeperCluster.getDescription());
                    newZooKeeperClusterKeyToZooKeeperClusterMap.put(zooKeeperClusterKey, zooKeeperCluster);
                } else {
                    iterator.remove();
                    closeZooKeeperClusterClient(zooKeeperCluster);
                }
            }
        }
    }

    private void connectZooKeeperCluster(Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap) {
        Iterator<Entry<String, ZooKeeperCluster>> iterator = zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, ZooKeeperCluster> next = iterator.next();
            ZooKeeperCluster zooKeeperCluster = next.getValue();
            ZooKeeper zooKeeper = zooKeeperCluster.getZooKeeper();
            if (zooKeeper == null) {
                connectZooKeeperCluster(zooKeeperCluster);
                if (!zooKeeperCluster.isOffline()) {
                    //...
                }
            }
        }
    }

    /**
     * 关闭迁移了的NameSpace
     *
     * @param zooKeeperClusterKey
     * @param nameSpaceZooKeeperClusterMappingDtoList 以这个为参考标准
     * @param nameSpaceConfigurationList
     */
    private void closeMigrationNameSpace(String zooKeeperClusterKey, List<NameSpaceZooKeeperClusterMappingDto> nameSpaceZooKeeperClusterMappingDtoList, List<NameSpaceConfiguration> nameSpaceConfigurationList) {
        if (nameSpaceConfigurationList == null) {
            return;
        }

        Iterator<NameSpaceConfiguration> iterator = nameSpaceConfigurationList.iterator();
        while (iterator.hasNext()) {
            NameSpaceConfiguration nameSpaceConfiguration = iterator.next();
            String nameSpace = nameSpaceConfiguration.getNameSpace();
            boolean include = false;
            if (nameSpaceZooKeeperClusterMappingDtoList != null) {
                for (NameSpaceZooKeeperClusterMappingDto dto : nameSpaceZooKeeperClusterMappingDtoList) {
                    if (nameSpace.equals(dto.getNameSpace())) {
                        include = true;
                        break;
                    }
                }
            }
            if (!include) {
                synchronized (getNameSpaceLock(nameSpace)) {
                    iterator.remove();
                    closeNameSpaceClient(nameSpace);
                    log.info("closed the moved namespace info, namespace is {}, old zkClusterKey is {}", nameSpace, zooKeeperClusterKey);
                }
            }
        }
    }

    private void initNameSpace(List<String> onlineNameSpaceList,//
                               ZooKeeperCluster zooKeeperCluster,//
                               List<NameSpaceZooKeeperClusterMappingDto> nameSpaceZooKeeperClusterMappingDtoList,//
                               List<NameSpaceConfiguration> nameSpaceConfigurationList) {//
        if (nameSpaceZooKeeperClusterMappingDtoList == null || zooKeeperCluster.isOffline()) {
            return;
        }
        ZooKeeper zooKeeper = zooKeeperCluster.getZooKeeper();
        List<NameSpaceConfiguration> newNameSpaceConfigurationList = new ArrayList<NameSpaceConfiguration>();
        try {
            for (NameSpaceZooKeeperClusterMappingDto dto : nameSpaceZooKeeperClusterMappingDtoList) {
                String nameSpace = dto.getNameSpace();
                // 过滤$SaturnSelf
                if (SelfNodePath.ROOT_NAME.equals(nameSpace)) {
                    continue;
                }
                // 如果这个标记为true，意味是新域，或者是迁移过来的域
                boolean isNamespaceNotIncludeInOriginRegCenerConfList = false;
                if (isNameSpaceNotIncludeInNameSpaceConfigurationList(nameSpace, nameSpaceConfigurationList)) {
                    // 对于新添加的域，需要初始化一些znode
                    initNameSpaceZooKeeperNodeIfNecessary(nameSpace, zooKeeper);
                    isNamespaceNotIncludeInOriginRegCenerConfList = true;
                }

                try {
                    if (isNamespaceNotIncludeInOriginRegCenerConfList) {
                        NameSpaceConfiguration nameSpaceConfiguration = new NameSpaceConfiguration(zooKeeperCluster.getZooKeeperAddress(), nameSpace);
                        nameSpaceConfiguration.setZooKeeperClusterKey(zooKeeperCluster.getZooKeeperClusterKey());
                        nameSpaceConfiguration.setAlias(zooKeeperCluster.getAlias());
                        newNameSpaceConfigurationList.add(nameSpaceConfiguration);
                        if (!onlineNameSpaceList.contains(nameSpace)) {
                            onlineNameSpaceList.add(nameSpace);
                        }
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        int oldSize = nameSpaceConfigurationList != null ? nameSpaceConfigurationList.size() : 0;
        if (oldSize != newNameSpaceConfigurationList.size()) {
            log.info("Zkcluster [{}] namespace size change from {} to {}", zooKeeperCluster.getZooKeeperClusterKey(), oldSize, newNameSpaceConfigurationList.size());
        }

        zooKeeperCluster.setNameSpaceConfigurationList(newNameSpaceConfigurationList);
    }

    private boolean isNameSpaceNotIncludeInNameSpaceConfigurationList(String namespace, List<NameSpaceConfiguration> nameSpaceConfigurationList) {
        if (nameSpaceConfigurationList == null || nameSpaceConfigurationList.isEmpty()) {
            return true;
        }
        for (NameSpaceConfiguration nameSpaceConfiguration : nameSpaceConfigurationList) {
            if (namespace.equals(nameSpaceConfiguration.getNameSpace())) {
                return false;
            }
        }
        return true;
    }

    /**
     * 创建或者移除namespaceShardingManager.
     */
    private void refreshNameSpaceToTaskManagerMap() {
        Iterator<Entry<String, ZooKeeperCluster>> iterator = zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, ZooKeeperCluster> next = iterator.next();
            ZooKeeperCluster zooKeeperCluster = next.getValue();
            List<NameSpaceConfiguration> namespaceConfigurationList = zooKeeperCluster.getNameSpaceConfigurationList();
            if (namespaceConfigurationList == null) {
                continue;
            }
            for (NameSpaceConfiguration nameSpaceConfiguration : namespaceConfigurationList) {
                String nameSpace = nameSpaceConfiguration.getNameSpace();
                if (!nameSpaceToTaskManagerMap.containsKey(nameSpace)) {
                    if (isZooKeeperClusterCanBeComputed(nameSpaceConfiguration.getZooKeeperClusterKey())) {
                        createNameSpaceTaskManager(nameSpaceConfiguration);
                    }
                } else {
                    TaskManager taskManager = nameSpaceToTaskManagerMap.get(nameSpace);
                    if (!isZooKeeperClusterCanBeComputed(nameSpaceConfiguration.getZooKeeperClusterKey())) {
                        taskManager.stopWithCurator();
                        nameSpaceToTaskManagerMap.remove(nameSpace);
                    }
                }
            }
        }
    }

    /**
     * 判断该集群是否能被本Console计算
     */
    private boolean isZooKeeperClusterCanBeComputed(String zooKeeperClusterKey) {
        return restrictComputeZkClusterKeys.contains(zooKeeperClusterKey);
    }

    private void createNameSpaceTaskManager(NameSpaceConfiguration nameSpaceConfiguration) {
        try {
            String nameSpace = nameSpaceConfiguration.getNameSpace();
            String digest = nameSpaceConfiguration.getDigest();
            CuratorFramework curatorFramework = zooKeeperRepository.connect(nameSpaceConfiguration.getZooKeeperAddressList(), nameSpace, digest);
            if (curatorFramework == null) {
                log.warn("fail to connect to zk during create NamespaceShardingManager");
                return;
            }

            TaskManager taskManager = null;
            try {
                taskManager = new TaskManager(curatorFramework, nameSpace, generateShardingLeadershipHostValue());
                taskManager.start();
                if (nameSpaceToTaskManagerMap.putIfAbsent(nameSpace, taskManager) != null) {
                    // 已经存在，则关闭当前的client
                    try {
                        taskManager.stopWithCurator();
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }
                } else {
                    log.info("Done starting NamespaceShardingManager {}", nameSpace);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                if (taskManager != null) {
                    try {
                        taskManager.stop();
                    } catch (Exception e2) {
                        log.error(e2.getMessage(), e2);
                    }
                }
                curatorFramework.close();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private String generateShardingLeadershipHostValue() {
        return LocalHostService.cachedIpAddress + "-" + UUID.randomUUID();
    }

    private void initLocalRefreshThreadPool() {
        localRefreshThreadPool = Executors.newSingleThreadExecutor(new ConsoleThreadFactory("refresh-RegCenter-thread", false));
    }

    private void startLocalRefreshTimer() {
        localRefreshTimer = new Timer("refresh-RegCenter-timer", true);
        // 每隔5分钟刷新一次
        localRefreshTimer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                try {
                    localRefreshThreadPool.submit(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                localRefresh();
                            } catch (Exception e) {
                                log.error(e.getMessage(), e);
                            }
                        }
                    });
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }, 1000L * 60 * 5, 1000L * 60 * 5);
    }

    private void startLocalRefreshIfNecessaryTimer() {
        localRefreshIfNecessaryTimer = new Timer("refresh-RegCenter-if-necessary-timer", true);
        localRefreshIfNecessaryTimer.schedule(new TimerTask() {

            private String lastUuid = null;

            @Override
            public void run() {
                try {
                    String uuid = null;//systemConfigurationService.getValueDirectly(SystemConfigProperties.REFRESH_REGISTRY_CENTER_UUID);
                    if (StringUtils.isBlank(uuid)) {
                        notifyRefresh();
                    } else if (!uuid.equals(lastUuid)) {
                        lastUuid = uuid;
                        localRefreshThreadPool.submit(new Runnable() {
                            @Override
                            public void run() {
                                try {
                                    localRefresh();
                                } catch (Exception e) {
                                    log.error(e.getMessage(), e);
                                }
                            }
                        });
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }, 1000, 1000);
    }

    /**
     * 从数据库获取基础信息
     *
     * @return
     */
    private Map<String, ZooKeeperCluster> getZooKeeperClusterKeyToZooKeeperClusterMap() {
        LinkedHashMap<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = new LinkedHashMap<String, ZooKeeperCluster>();
        List<ZooKeeperClusterDto> dtoList = zooKeeperClusterDtoService.getZooKeeperClusterInfoList();
        if (dtoList != null) {
            for (ZooKeeperClusterDto dto : dtoList) {
                ZooKeeperCluster zooKeeperCluster = new ZooKeeperCluster();
                zooKeeperCluster.setZooKeeperClusterKey(dto.getZooKeeperClusterKey());
                zooKeeperCluster.setAlias(dto.getAlias());
                zooKeeperCluster.setZooKeeperAddress(dto.getConnectString());
                zooKeeperCluster.setDescription(dto.getDescription());
                zooKeeperClusterKeyToZooKeeperClusterMap.put(dto.getZooKeeperClusterKey(), zooKeeperCluster);
            }
        }
        return zooKeeperClusterKeyToZooKeeperClusterMap;
    }


    private Object getNameSpaceLock(String nameSpace) {
        Object lock = nameSpaceLock.get(nameSpace);
        if (lock == null) {
            lock = new Object();
            Object pre = nameSpaceLock.putIfAbsent(nameSpace, lock);
            if (pre != null) {
                lock = pre;
            }
        }
        return lock;
    }

    private void closeNameSpaceClient(String nameSpace) {
        try {
            NameSpaceClient nameSpaceClient = nameSpaceToNameSpaceClientMap.remove(nameSpace);
            if (nameSpaceClient != null) {
                nameSpaceClient.close();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        try {
            TaskManager taskAllotManager = nameSpaceToTaskManagerMap.remove(nameSpace);
            if (taskAllotManager != null) {
                taskAllotManager.stopWithCurator();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


    /**
     * Close dashboardLeaderTreeCache, registryCenterClient, namespaceShardingListenerManager with this zkCluster
     */
    private void closeZooKeeperClusterClient(ZooKeeperCluster zooKeeperCluster) {
        try {
            List<NameSpaceConfiguration> namespaceConfigurationList = zooKeeperCluster.getNameSpaceConfigurationList();
            if (namespaceConfigurationList != null) {
                for (NameSpaceConfiguration nameSpaceConfiguration : namespaceConfigurationList) {
                    String nameSpace = nameSpaceConfiguration.getNameSpace();
                    synchronized (getNameSpaceLock(nameSpace)) {
                        closeNameSpaceClient(nameSpace);
                    }
                }
            }
            if (zooKeeperCluster.getConnectionStateListener() != null) {
                zooKeeperCluster.getConnectionStateListener().shutdownNowUntilTerminated();
                zooKeeperCluster.setConnectionStateListener(null);
            }
            if (zooKeeperCluster.getZooKeeper() != null && zooKeeperCluster.getZooKeeper().getCuratorFramework() != null) {
                zooKeeperCluster.getZooKeeper().getCuratorFramework().close();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void connectZooKeeperCluster(final ZooKeeperCluster zooKeeperCluster) {
        String zooKeeperAddress = zooKeeperCluster.getZooKeeperAddress();
        try {
            final CuratorFramework curatorFramework = zooKeeperRepository.connect(zooKeeperAddress, null, zooKeeperCluster.getDigest());
            if (curatorFramework == null) {
                log.error("found an offline zkCluster, zkAddr is {}", zooKeeperAddress);
                zooKeeperCluster.setZooKeeper(null);
                zooKeeperCluster.setConnectionStateListener(null);
                zooKeeperCluster.setOffline(true);
            } else {
                ConnectionStateListener connectionStateListener = new ConnectionStateListener("zk-connectionListener-thread-for-zkCluster-" + zooKeeperCluster.getAlias()) {
                    @Override
                    public void stop() {
                        zooKeeperCluster.setOffline(true);
                    }

                    @Override
                    public void restart() {
                        try {
                            zooKeeperCluster.setOffline(false);
                        } catch (Exception e) {
                            log.error(e.getMessage(), e);
                        }
                    }
                };
                zooKeeperCluster.setZooKeeper(zooKeeperRepository.getZooKeeper(curatorFramework));
                zooKeeperCluster.setConnectionStateListener(connectionStateListener);
                zooKeeperCluster.setOffline(false);
                curatorFramework.getConnectionStateListenable().addListener(connectionStateListener);
            }
        } catch (Exception e) {
            log.error("found an offline zkCluster, zkAddr is {}", zooKeeperAddress);
            log.error(e.getMessage(), e);
            zooKeeperCluster.setZooKeeper(null);
            zooKeeperCluster.setConnectionStateListener(null);
            zooKeeperCluster.setOffline(true);
        }
    }

    protected void initNameSpaceZooKeeperNodeIfNecessary(String nameSpace, ZooKeeper zooKeeper) {
        try {
            String taskIdsNodePath = "/" + nameSpace + TaskNodePath.getTasksNodePath();
            if (!zooKeeper.checkExists(taskIdsNodePath)) {
                zooKeeper.create(taskIdsNodePath);
            }
            String executorIdsNodePath = "/" + nameSpace + ExecutorNodePath.get$ExecutorNodePath();
            if (!zooKeeper.checkExists(executorIdsNodePath)) {
                zooKeeper.create(executorIdsNodePath);
            }

        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


    @Override
    public NameSpaceConfiguration findConfigByNameSpace(String nameSpace) {
        if (Strings.isNullOrEmpty(nameSpace)) {
            return null;
        }
        Collection<ZooKeeperCluster> zooKeeperClusterCollection = zooKeeperClusterKeyToZooKeeperClusterMap.values();
        for (ZooKeeperCluster zooKeeperCluster : zooKeeperClusterCollection) {
            for (NameSpaceConfiguration nameSpaceConfiguration : zooKeeperCluster.getNameSpaceConfigurationList()) {
                if (nameSpaceConfiguration != null && nameSpace.equals(nameSpaceConfiguration.getNameSpace())) {
                    return nameSpaceConfiguration;
                }
            }
        }
        return null;
    }

    @Override
    public void notifyRefresh() throws ConsoleException {
        //DB更新 UUID
    }


    @Override
    public List<ZooKeeperCluster> getZooKeeperClusterList() {
        return new ArrayList<>(zooKeeperClusterKeyToZooKeeperClusterMap.values());
    }

    private List<String> getZooKeeperClusterKeyList() {
        Collection<ZooKeeperCluster> zooKeeperClusterList = getZooKeeperClusterList();
        List<String> zooKeeperClusterKeyList = Lists.newArrayList();
        for (ZooKeeperCluster zooKeeperCluster : zooKeeperClusterList) {
            zooKeeperClusterKeyList.add(zooKeeperCluster.getZooKeeperClusterKey());
        }
        return zooKeeperClusterKeyList;
    }


    @Override
    public ZooKeeper getZooKeeper(String nameSpace) throws ConsoleException {
        ZooKeeper zooKeeper = null;
        try {
            NameSpaceConfiguration nameSpaceConfiguration = findConfigByNameSpace(nameSpace);
            if (nameSpaceConfiguration == null) {
                throw new ConsoleException("find registryCenterConfiguration failed");
            }
            String zooKeeperAddressList = nameSpaceConfiguration.getZooKeeperAddressList();
            String digest = nameSpaceConfiguration.getDigest();
            synchronized (getNameSpaceLock(nameSpace)) {
                if (!nameSpaceToNameSpaceClientMap.containsKey(nameSpace)) {
                    final NameSpaceClient nameSpaceClient = new NameSpaceClient();
                    nameSpaceClient.setNameSpace(nameSpace);
                    nameSpaceClient.setZooKeeperAddress(zooKeeperAddressList);
                    CuratorFramework curatorFramework = zooKeeperRepository.connect(zooKeeperAddressList, nameSpace, digest);
                    if (curatorFramework != null) {
                        nameSpaceClient.setConnected(curatorFramework.getZookeeperClient().isConnected());
                        nameSpaceClient.setCuratorFramework(curatorFramework);
                        nameSpaceToNameSpaceClientMap.put(nameSpace, nameSpaceClient);
                        zooKeeper = zooKeeperRepository.getZooKeeper(curatorFramework);
                    }
                } else {
                    NameSpaceClient nameSpaceClient = nameSpaceToNameSpaceClientMap.get(nameSpace);
                    if (nameSpaceClient != null) {
                        CuratorFramework curatorFramework = nameSpaceClient.getCuratorFramework();
                        if (curatorFramework != null) {
                            nameSpaceClient.setConnected(curatorFramework.getZookeeperClient().isConnected());
                            zooKeeper = zooKeeperRepository.getZooKeeper(curatorFramework);
                        }
                    }
                }
            }
        } catch (ConsoleException e) {
            throw e;
        } catch (Exception e) {
            throw new ConsoleException(e);
        }
        if (zooKeeper == null) {
            throw new ConsoleException("Connect zookeeper failed");
        }
        return zooKeeper;
    }


}
